package com.alibaba.datax.core.transport.transformer;

import com.alibaba.datax.common.element.Column;
import com.alibaba.datax.common.element.Record;
import com.alibaba.datax.common.element.StringColumn;
import com.alibaba.datax.common.exception.DataXException;
import com.alibaba.datax.transformer.Transformer;
import groovy.util.logging.Log4j;

import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;

/**
 *
 * {
 *     "name": "dx_dictionary",
 *     "parameter": {
 *         "columnIndex": 5,
 *         "paras": [
 *             "0",
 *             {
 *                 "10": "0",
 *                 "11": "1",
 *                 "11": "1",
 *             }
 *         ]
 *     }
 * }
 *
 * @author 张陶
 *
 */
public class DictionaryTransformer extends Transformer {

    private ThreadLocal<Map<Integer, Map<String, String>>> dictMapCache = new ThreadLocal<>();

    public DictionaryTransformer() {
        setTransformerName("dx_dictionary");
    }

    @Override
    public Record evaluate(Record record, Object... paras) {
        try {
            if (paras.length != 4) {
                throw new RuntimeException("dx_dictionary paras must be 4");
            }
            Integer columnIndex = (Integer) paras[0];
            String defaultValue = (String) paras[1];
            Map<Integer, Map<String, String>> dictTypeMap = dictMapCache.get();
            if (dictTypeMap == null) {
                dictTypeMap = new HashMap<>();
                dictMapCache.set(dictTypeMap);
            }
            Map<String, String> dictMap = dictTypeMap.get(columnIndex);
            if (dictMap == null) {
                String[] oriDict = ((String) paras[2]).split(",");
                String[] newDict = ((String) paras[3]).split(",");
                if (oriDict.length != newDict.length) {
                    throw new RuntimeException("oriDict.length must equal newDict.length ");
                }
                dictMap = new HashMap<>();
                for (int i=0; i<oriDict.length; i++) {
                    dictMap.put(oriDict[i], newDict[i]);
                }
                dictTypeMap.put(columnIndex, dictMap);
            }
            Column column = record.getColumn(columnIndex);
            String oriValue = column.asString();
            String newValue = dictMap.getOrDefault(oriValue, defaultValue);
            record.setColumn(columnIndex, new StringColumn(newValue));
        } catch (Exception e) {
            throw DataXException.asDataXException(TransformerErrorCode.TRANSFORMER_ILLEGAL_PARAMETER, "paras:" + Arrays.asList(paras).toString() + " => " + e.getMessage());
        }

        return record;
    }
}
